package com.dgw.springbootkafka.receiver;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;


@Component
public class KafkaReceiver {
    private static final Logger log = LoggerFactory.getLogger(KafkaReceiver.class);

    //@KafkaListener(topics = "goods")
    @KafkaListener(topics = "test-syn", groupId = "test-1")
    public void send(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object messge = kafkaMessage.get();
            log.info("【KafkaListener监听到消息】1" + messge);
        }
    }

    @KafkaListener(topics = "test-syn", groupId = "test-2")
    public void send2(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object messge = kafkaMessage.get();
            log.info("【KafkaListener监听到消息】2" + messge);
        }
    }

}
